package cn.org.intelli.zjgflink.controller;

import cn.org.intelli.zjgflink.domain.BackBean;
import cn.org.intelli.zjgflink.util.FlinkJobDeploy;
import cn.org.intelli.zjgflink.util.FlinkJobStop;
import cn.org.intelli.zjgflink.util.Uid;
import org.apache.flink.runtime.checkpoint.Checkpoints;
import org.apache.flink.runtime.checkpoint.metadata.CheckpointMetadata;
import org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorageAccess;
import org.apache.flink.runtime.state.filesystem.FsCompletedCheckpointStorageLocation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.repository.query.Param;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.io.DataInputStream;
import java.io.IOException;

@RestController
public class TestController {
    @Autowired
    private FlinkJobDeploy flinkJobDeploy;

    @Autowired
    private FlinkJobStop flinkJobStop;

    @GetMapping({"/readMetadata"})
    public BackBean<String> readMetaDataFile(@Param("savePath") String savePath) throws Exception {
        FsCompletedCheckpointStorageLocation fsCompletedCheckpointStorageLocation = AbstractFsCheckpointStorageAccess.resolveCheckpointPointer(savePath);
        try (DataInputStream stream = new DataInputStream(fsCompletedCheckpointStorageLocation.getMetadataHandle().openInputStream())) {
            CheckpointMetadata metadata = Checkpoints.loadCheckpointMetadata(stream,
                    Thread.currentThread().getContextClassLoader(), savePath);
            System.out.println(metadata);
            System.out.println(metadata.getCheckpointId());
            System.out.println(metadata.getMasterStates());
            System.out.println(metadata.getOperatorStates());
            System.out.println(metadata.getMasterStates().size());
            return BackBean.<String>builder().status(true).data(metadata.toString()).build();
        } catch (IOException e) {
            e.printStackTrace();
            return BackBean.<String>builder().status(false).data(null).build();
        }
    }

    @GetMapping({"/test"})
    public String test() {
        try {
            String flinkJobId = Uid.generateShortUuid();
            String flinkJobName = "flinkJobForTest";
            Integer slotNumber = Integer.valueOf(3);
            String mainFunction = "org.example.ExceptionStreamDemo";
            String jarFileName = "ZJGStream-7.0-SNAPSHOT.jar";
            String savePointDir = null;
            String[] arrayOfString = new String[0];
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success";
    }

    @GetMapping({"/test2"})
    public String test2() {
        try {
            String appId = "application_1736500205863_0054";
            String jobId = "wZlBghqV";
            this.flinkJobStop.stopFlinkJob(appId, jobId);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "success";
    }
}
